package com.hzgj.soa.protocol.pb.handler;

import com.baidu.bjf.remoting.protobuf.Codec;
import com.baidu.bjf.remoting.protobuf.ProtobufProxy;
import com.hzgj.bcl.soa.net.Decoder;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.soa.protocol.pb.data.RequestMessage;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;

import java.net.ProtocolException;
import java.nio.charset.StandardCharsets;
import java.util.List;

import static com.hzgj.soa.protocol.pb.handler.PbConstants.PB_PROTOCOL;

public class PbDecoder extends Decoder {
    private State state = State.HEADER;
    private int length;
    private int message_type;
    private static Codec<RequestMessage> requestCodec = ProtobufProxy.create(RequestMessage.class);

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        decode(in, out);
    }

    private void decode(ByteBuf buf, List<Object> out) throws Exception {
        // "RESPONSE_MAGIC" + ' ' + [length] + "\r\n" + [data]
        switch (state) {
            case HEADER:
                if (buf.readableBytes() < PbConstants.responseMagicLength()) {
                    return;
                } else {
                    int magic = buf.readInt();
                    if (magic != PbConstants.RESPONSE_MAGIC && magic != PbConstants.REQUEST_MAGIC) {
                        throw new ProtocolException("expect [protobuf protocol]");
                    }
                    if (magic == PbConstants.REQUEST_MAGIC) {
                        message_type = Message.REQ;
                    } else {
                        message_type = Message.RSP;
                    }
                    state = State.LENGTH;
                    decode(buf, out);
                }
                break;
            case LENGTH:
                if (buf.readableBytes() == 0) {
                    return;
                } else {
                    int number = buf.bytesBefore((byte) '\r');
                    if (number == -1) {
                        return;
                    } else {
                        byte[] bytes = new byte[number];
                        buf.readBytes(bytes);
                        length = Integer.parseInt(new String(bytes, StandardCharsets.US_ASCII));
                        state = State.CRLF;
                        decode(buf, out);
                    }
                }
                break;
            case CRLF:
                if (buf.readableBytes() < 2) {
                    return;
                } else {
                    if (buf.readByte() != '\r' || buf.readByte() != '\n') {
                        throw new ProtocolException("expect [\\r\\n]");
                    }
                    state = State.BODY;
                    decode(buf, out);
                }
                break;
            case BODY:
                if (buf.readableBytes() < length) {
                    return;
                } else {
                    byte[] bytes = new byte[length];
                    buf.readBytes(bytes);
                    Message message = new Message();
                    if (message_type == Message.REQ) {
                        RequestMessage requestMessage = requestCodec.decode(bytes);
                        message.putHeader(Constants.MESSAGE_TYPE, message_type)
                                .putHeader(Constants.MESSAGE_HEADER_PROTOCOL, PB_PROTOCOL)
                                .putHeader(Constants.MESSAGE_SID, requestMessage.messageId)
                                .putHeader(Constants.MESSAGE_HEADER_SERIALIZATION_BYPASS, true)
                                .setPayload(requestMessage);
                    } else {
                        //SID包含在ResponseMessage中，在序列化中再放置到头部
                        message.putHeader(Constants.MESSAGE_TYPE, message_type)
                                .putHeader(Constants.MESSAGE_HEADER_PROTOCOL, PB_PROTOCOL)
                                .setPayload(bytes);
                    }
                    out.add(message);
                    state = State.HEADER;
                    if (buf.readableBytes() > 0) {
                        decode(buf, out);
                    }
                }
                break;
        }
    }

    @Override
    public String getName() {
        return PB_PROTOCOL;
    }

    enum State {
        HEADER, LENGTH, CRLF, BODY
    }
}